package com.hbase;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;



import java.io.IOException;
import java.util.Date;
import java.util.List;

/**
 * @author zhaoyi
 * @date 2020/9/27 10:22
 */
public class HbaseToHdfs {

    public static void main(String[] args) throws IOException {

        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "blueview-s020,blueview-s021,blueview-s022,blueview-s023,blueview-s024");// zookeeper地址
        config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口

        /*config.setInt("hbase.rpc.timeout", 70000);
        config.setInt("hbase.client.operation.timeout", 70000);
        config.setInt("hbase.client.scanner.timeout.period", 70000);*/


        Connection connection = ConnectionFactory.createConnection(config);
        Table table = connection.getTable(TableName.valueOf("wbpost"));
        Scan scan = new Scan();
        //scan.setStartRow(Bytes.toBytes("1DHUQ-1585242180000-IAsuAglOP"));
        scan.setStartRow(Bytes.toBytes("2N86D-1580829960000-IsH3Ujofx"));
        scan.setStopRow(Bytes.toBytes("2O79T-1582891860000-Iwk2Hu0rm"));

        List<String> columns = Lists.newArrayList("wbId", "createdAt", "userId", "userUrn", "postUrn", "content");
        columns.forEach(col -> {
            scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes(col));
        });

        DateTime dateTime = DateUtil.parse("202010", "yyyyMM");
        long begin = DateUtil.beginOfMonth(dateTime).getTime();
        long end = DateUtil.endOfMonth(dateTime).getTime();



        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("createdAt"),
                CompareFilter.CompareOp.GREATER_OR_EQUAL, Bytes.toBytes(begin));
        filter1.setFilterIfMissing(true);

        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("createdAt"),
                CompareFilter.CompareOp.LESS_OR_EQUAL, Bytes.toBytes(end));
        filter2.setFilterIfMissing(true);

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        filterList.addFilter(filter1);
        filterList.addFilter(filter2);
        //scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);
        int i = 0;
        for (Result result : scanner) {
            i++;
            String content = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("content")));
            String rowKey = Bytes.toString(result.getRow());
            long createdTime = Bytes.toLong(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("createdAt")));

            Date date = new Date(createdTime);
            System.out.println(DateUtil.format(date,"yyyy-MM-dd"));
            if (i >= 100) {
                return;
            }
        }


        table.close();
        connection.close();


    }
}
